package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.beans.TrafficHomeDetailPageViewBean;
import com.atguigu.realtime.utils.DateFormatUtil;
import com.atguigu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author: 洛尘
 * @since: 2023-10-12 21:07
 * @description: 首页，详情页独立访客统计
 **/
public class DwsTrafficHomeDetailPageViewWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(5000L);
        String topic="dwd_traffic_page_log";
        String groupId="dws_traffic_home_detail_group";

        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);

        SingleOutputStreamOperator<String> kafkaDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.map(JSON::parseObject);

        SingleOutputStreamOperator<JSONObject> filterDS = jsonDS.filter(
                new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String pageId = jsonObject.getJSONObject("page").getString("page_id");
                        return "home".equals(pageId) || "good_detail".equals(pageId);
                    }
                }
        );
        filterDS.print(">>>>>");
        SingleOutputStreamOperator<JSONObject> waDS = filterDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts");
                                    }
                                }
                        )
        );
        KeyedStream<JSONObject, String> keyDS =
                waDS.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));

        keyDS.process(
                new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {
                    private ValueState<String> homeLastVisitDateState;
                    private ValueState<String> detailLastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<String> homeLastVisitDateStateDescriptor =
                                new ValueStateDescriptor<String>("homeLastVisitDateStateDescriptor",String.class);
                        homeLastVisitDateStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        homeLastVisitDateState=getRuntimeContext().getState(homeLastVisitDateStateDescriptor);

                        ValueStateDescriptor<String> detailLastVisitDateState =
                                new ValueStateDescriptor<>("detailLastVisitDateState", String.class);
                        detailLastVisitDateState.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        this.detailLastVisitDateState =getRuntimeContext().getState(detailLastVisitDateState);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {
                        String page = jsonObject.getJSONObject("page").getString("page_id");
                        Long ts = jsonObject.getLong("ts");
                        String curVisitDate = DateFormatUtil.toDate(ts);

                        Long homeUvCt=0L;
                        Long detailUvCt=0L;
                        if ("home".equals(page)){
                            //判断是不是及独立访客
                            String homeLastVisitDate = homeLastVisitDateState.value();
                            if (StringUtils.isEmpty(homeLastVisitDate)||!homeLastVisitDate.equals(curVisitDate)){
                                homeUvCt=1L;
                                homeLastVisitDateState.update(curVisitDate);
                            }
                        }
                    }
                }
        );
        env.execute();

    }
}